package com.ang.mqtt.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import javax.annotation.Resource;

@Configuration
public class MqttListenerConfig {

    public static final String CHANNEL_NAME_IN = "mqttInboundChannel";

//    private static String ClientId = "test/consumer";
//    private static String ListenTopic = "towerTest";

    @Autowired
    MqttPahoClientFactory mqttClientFactory;

    @Resource
    MqttProperties mqttProperties;

    /**
     * MQTT消息通道（消费者）
     */
    @Bean(name = CHANNEL_NAME_IN)
    public MessageChannel mqttInboundChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT消息订阅绑定（消费者）
     */
    @Bean
    public MessageProducer mqttInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId(),
                mqttClientFactory, mqttProperties.getSubTopic());

        //设置超时时间
        adapter.setCompletionTimeout(5000);
        //设置默认的消息转换类
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        //设置入站管道
        adapter.setOutputChannel(mqttInboundChannel());

        return adapter;
    }

    /**
     * MQTT消息监听器（消费者）
     * MessageHandler: org.springframework:spring-messaging
     */
    @Bean
    @ServiceActivator(inputChannel = CHANNEL_NAME_IN)
    public MessageHandler handlerTest() {

        return message -> {
            try {
                String string = message.getPayload().toString();
                System.out.println("接收到消息：" + string);
            } catch (MessagingException e) {
                e.printStackTrace();
                //logger.info(e.getMessage());
            }
        };
    }

}

